package com.youku.collector;

import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import com.youku.collector.util.Assert;

public class Producer implements Task<Record>, Counter {

	private BlockingQueue<Record> queue;

	private File root;

	private FileFilter filter;

	private AtomicInteger count = new AtomicInteger(0);

	public Producer(BlockingQueue<Record> queue, File root, FileFilter filter) {
		super();
		Assert.notNull(queue);
		Assert.notNull(root);
		Assert.notNull(filter);
		this.queue = queue;
		this.root = root;
		this.filter = filter;
		
	}

	@Override
	public void run() {
		try {
			addStart();
			crawl(root);
			addEnd();
			System.out.println("collector count is " + count);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	private void addEnd() throws InterruptedException {
		queue.put(new Record(null, null, "end", null, 0));
	}

	private void addStart() throws InterruptedException {
		queue.put(new Record(null, null, "start", null, 0));
	}

	private void crawl(File file) throws InterruptedException {
		File[] children = file.listFiles(filter);
		if (children != null) {
			for (File child : children) {
				if (child.isDirectory()) {
					crawl(child);
				} else {
					incr();
					queue.put(
							new Record(file.getParentFile().getName(), file.getName(), "consumer", child.getName(), 0));
				}
			}
		}
	}

	@Override
	public void incr() {
		count.incrementAndGet();
	}

	@Override
	public int count() {
		return count.get();
	}

	@Override
	public BlockingQueue<Record> getQueue() {
		return queue;
	}

	@Override
	public void put(Record element) throws InterruptedException {
		queue.put(element);
	}

	@Override
	public Record take() throws InterruptedException {
		throw new UnsupportedMethodException("生产者不支持take方法");
	}

}
